package com.fyt.reachgo.consummer;

import com.fyt.reachgo.config.RabbitMQConfig;
import com.fyt.reachgo.domain.UserFavorite;
import com.fyt.reachgo.service.UserFavoriteService;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.time.Duration;

import static com.fyt.reachgo.service.impl.UserFavoriteServiceImpl.USER_FAVORITES_LIST_KEY;

@Component
@Slf4j
@RequiredArgsConstructor
public class CollectConsumer {
    private static final String FAVORITE_IDEMPOTENT_KEY_PREFIX = "favorite:";
    private static final Duration IDEMPOTENT_KEY_TTL = Duration.ofMinutes(5);

    private final UserFavoriteService userFavoriteService;
    private final StringRedisTemplate stringRedisTemplate;

    @RabbitListener(
            queues = RabbitMQConfig.USER_FAVORITE_QUEUE,
            ackMode = "MANUAL",
            concurrency = "3-5" // 根据实际情况调整并发数
    )
    @Transactional(rollbackFor = Exception.class)
    public void handleCollectAdd(UserFavorite userFavorite,
                                 Channel channel,
                                 @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            log.info("[收藏消息] 开始处理收藏消息, 用户ID: {}, 内容ID: {}, 内容类型: {}",
                    userFavorite.getUserId(),
                    userFavorite.getTargetId(),
                    userFavorite.getTargetType());

            // 幂等性检查
            if (!checkAndSetIdempotent(userFavorite)) {
                log.warn("[收藏消息] 重复消息已忽略, 用户ID: {}, 内容ID: {}",
                        userFavorite.getUserId(),
                        userFavorite.getTargetId());
                channel.basicAck(tag, false);
                return;
            }

            // 处理收藏逻辑
            userFavoriteService.save(userFavorite);
            // 删除缓存
            stringRedisTemplate.delete(USER_FAVORITES_LIST_KEY + userFavorite.getUserId());

            // 确认消息
            channel.basicAck(tag, false);
            log.info("[收藏消息] 处理成功, 用户ID: {}, 内容ID: {}",
                    userFavorite.getUserId(),
                    userFavorite.getTargetId());
        } catch (Exception e) {
            log.error("[收藏消息] 处理失败, 用户ID: {}, 内容ID: {}",
                    userFavorite.getUserId(),
                    userFavorite.getTargetId(),
                    e);
            try {
                // 处理失败，拒绝消息并重新入队
                channel.basicNack(tag, false, true);
            } catch (IOException ex) {
                log.error("[收藏消息] 消息NACK失败, 用户ID: {}, 内容ID: {}",
                        userFavorite.getUserId(),
                        userFavorite.getTargetId(),
                        ex);
            }
            throw new RuntimeException(e); // 触发事务回滚
        }
    }

    /**
     * 幂等性检查
     */
    private boolean checkAndSetIdempotent(UserFavorite userFavorite) {
        String idempotentKey = FAVORITE_IDEMPOTENT_KEY_PREFIX
                + userFavorite.getUserId() + ":"
                + userFavorite.getTargetId() + ":"
                + userFavorite.getTargetType();
        return Boolean.TRUE.equals(stringRedisTemplate.opsForValue()
                .setIfAbsent(idempotentKey, "1", IDEMPOTENT_KEY_TTL));
    }
}
